package com.aaa.shuke2024.consumer;

import com.aaa.shuke2024.entity.AlarmHistory;
import com.aaa.shuke2024.entity.AlarmLimit;
import com.aaa.shuke2024.entity.Status;
import com.aaa.shuke2024.service.AlarmHistoryService;
import com.aaa.shuke2024.service.AlarmLimitService;
import com.aaa.shuke2024.service.StatusService;
import com.aaa.shuke2024.service.impl.MyEmailService;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.java.Log;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.Date;

/**
 * @author ：Teacher陈（86521760@qq.com）
 * @date ：Created in 2024/11/25 16:21
 * @description：我的消费者类
 * @modified By：
 * @version: 1.0
 */
@Component
@Log
public class MyConsumer {
    @Autowired
    private StatusService statusService;
    @Autowired
    private AlarmLimitService alarmLimitService;
    @Autowired
    private AlarmHistoryService alarmHistoryService;
    @Autowired
    private MyEmailService myEmailService;

    //KafkaListener监听器，作用是监听kafaka集群中的新消息
    @KafkaListener(topics = "chenjian")
    public void consumeMessage(ConsumerRecord record) {
        //获取kafka中的新消息
        Object value = record.value();
        System.out.println("通过监听器获取的最新消息：" + value.toString());
        //将获取的字符串转成对象
        ObjectMapper objectMapper = new ObjectMapper();
        Status status = null;
        try {
            status = objectMapper.readValue(value.toString(), Status.class);
            //将数据保存到mysql数据库中
            statusService.insert(status);
            //判断服务器健康指标是否正常，如果不正常发送邮件告警
            AlarmLimit alarmLimit = alarmLimitService.queryByIp(status.getServerIp());
            //获取告警的历史信息数据
            AlarmHistory alarmHistory = alarmHistoryService.queryByIp(status.getServerIp());
            //cpu
            if (status.getCpuUsage() > alarmLimit.getCpuLimit()) {
                String subject = "CPU负载告警";
                String content = "服务器" + status.getServerIp() + "cpu负载:" + status.getCpuUsage() + "超过警戒值:" + alarmLimit.getCpuLimit() + "请登录服务器处理";
                sendEmail(alarmLimit, alarmHistory, status,subject,content);
            }
            //内存
            if (status.getMemoryUsage() > alarmLimit.getMemoryLimit()) {
                String subject = "内存使用率告警";
                String content = "服务器" + status.getServerIp() + "内存使用率:" + status.getMemoryUsage() + "超过警戒值:" + alarmLimit.getMemoryLimit() + "请登录服务器处理";
                sendEmail(alarmLimit, alarmHistory, status,subject,content);
            }
            //硬盘
            if (status.getDiskUsage() > alarmLimit.getDiskLimit()) {
                String subject = "硬盘使用率告警";
                String content = "服务器" + status.getServerIp() + "硬盘使用率:" + status.getDiskUsage() + "超过警戒值:" + alarmLimit.getDiskLimit() + "请登录服务器处理";
                sendEmail(alarmLimit, alarmHistory, status,subject,content);
            }

        } catch (
                JsonProcessingException e) {
            throw new RuntimeException(e);
        }

    }

    public void sendEmail(AlarmLimit alarmLimit, AlarmHistory alarmHistory, Status status,String subject,String content) {
        //发邮件
        if (null != alarmHistory) {
            //发送邮箱,如果没有处理告警，30分钟之后再次发送
            long last = alarmHistory.getCreateTime().getTime();
            long now = System.currentTimeMillis();
            long diff = now - last;

            if (diff > 30 * 60 * 1000) {
                String emailAddress = alarmLimit.getEmailAddress();

                myEmailService.sendMail(emailAddress, subject, content);

                //存数据
                alarmHistory.setMessage(content);
                alarmHistory.setCreateTime(new Date());
                alarmHistory.setAlarmLimitId(alarmLimit.getId());
                alarmHistoryService.insert(alarmHistory);
            }
        } else {
            String emailAddress = alarmLimit.getEmailAddress();
            myEmailService.sendMail(emailAddress, subject, content);
            //存数据
            alarmHistory.setMessage(content);
            alarmHistory.setCreateTime(new Date());
            alarmHistory.setAlarmLimitId(alarmLimit.getId());
            alarmHistoryService.insert(alarmHistory);
        }

    }
}
